#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../common/threadpool.hpp"
#include "../common/msg.pb.h"
#include "../common/proto.pb.h"
#include "../common/logger.hpp"
#include "connection.hpp"
#include "consumer.hpp"
#include "host.hpp"

/**
 * @file broker.h
 * @brief 消息队列Broker服务器核心模块
 * 
 * 本模块实现消息队列Broker的核心服务器功能，包括：
 * 1. TCP网络通信服务
 * 2. AMQP协议命令处理分发
 * 3. 虚拟主机管理
 * 4. 消费者管理
 * 5. 连接与会话管理
 * 
 * 主要包含Server类，负责监听客户端连接，处理各类AMQP协议请求，
 * 并协调虚拟主机、连接管理器、消费者管理器等核心组件工作。
 */

namespace bitmq 
{
    #define DBFILE "/meta.db"     ///< 元数据数据库文件名
    #define HOSTNAME "MyVirtualHost" ///< 默认虚拟主机名称

    /**
     * @class Server
     * @brief 消息队列Broker核心服务器类
     * 
     * 负责启动TCP服务，处理客户端连接，分发AMQP协议命令，
     * 并管理虚拟主机、消费者、连接等核心资源。
     */
    class Server 
    {
        public:
            /// Protobuf消息智能指针类型
            typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
            
            /**
             * @brief 构造函数
             * @param port 服务器监听端口
             * @param basedir 虚拟主机数据存储基础目录
             * 
             * 初始化TCP服务器、协议编解码器、请求分发器、虚拟主机、
             * 消费者管理器、连接管理器和线程池。注册各类AMQP请求处理回调。
             */
            Server(int port, const std::string &basedir): _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), 
                "Server", muduo::net::TcpServer::kReusePort),
                _dispatcher(std::bind(&Server::onUnknownMessage, this, std::placeholders::_1, 
                    std::placeholders::_2, std::placeholders::_3)),
                _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                _virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
                _consumer_manager(std::make_shared<ConsumerManager>()),
                _connection_manager(std::make_shared<ConnectionManager>()),
                _threadpool(std::make_shared<threadpool>()){
                // 初始化所有队列的消费者管理结构
                QueueMap qm = _virtual_host->allQueues();
                for (auto &q : qm) 
                {
                    _consumer_manager->initQueueConsumer(q.first);
                }
                // 注册AMQP协议请求处理回调
                _dispatcher.registerMessageCallback<bitmq::openChannelRequest>(std::bind(&Server::onOpenChannel, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::closeChannelRequest>(std::bind(&Server::onCloseChannel, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::declareQueueRequest>(std::bind(&Server::onDeclareQueue, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::queueBindRequest>(std::bind(&Server::onQueueBind, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::queueUnBindRequest>(std::bind(&Server::onQueueUnBind, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::basicPublishRequest>(std::bind(&Server::onBasicPublish, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::basicAckRequest>(std::bind(&Server::onBasicAck, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::basicConsumeRequest>(std::bind(&Server::onBasicConsume, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _dispatcher.registerMessageCallback<bitmq::basicCancelRequest>(std::bind(&Server::onBasicCancel, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

                // 设置TCP服务器回调
                _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
            }
            
            /**
             * @brief 启动Broker服务器
             * 
             * 启动TCP监听服务并进入事件循环，开始处理客户端请求。
             */
            void start() 
            {
                _server.start();
                _baseloop.loop();
            }
            
        private:
            /**
             * @brief 处理打开信道请求
             * @param conn TCP连接对象
             * @param message openChannelRequest协议消息
             * @param 时间戳（未使用）
             */
            void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr)
                {
                    DLOG("打开信道时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                return mconn->openChannel(message);
            }
            
            /**
             * @brief 处理关闭信道请求
             * @param conn TCP连接对象
             * @param message closeChannelRequest协议消息
             * @param 时间戳（未使用）
             */
            void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("关闭信道时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                return mconn->closeChannel(message);
            }
            
            /**
             * @brief 处理声明交换机请求
             * @param conn TCP连接对象
             * @param message declareExchangeRequest协议消息
             * @param 时间戳（未使用）
             */
            void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("声明交换机时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("声明交换机时，没有找到信道！");
                    return;
                }
                return cp->declareExchange(message);
            }
            
            /**
             * @brief 处理删除交换机请求
             * @param conn TCP连接对象
             * @param message deleteExchangeRequest协议消息
             * @param 时间戳（未使用）
             */
            void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("删除交换机时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("删除交换机时，没有找到信道！");
                    return;
                }
                return cp->deleteExchange(message);
            }
            
            /**
             * @brief 处理声明队列请求
             * @param conn TCP连接对象
             * @param message declareQueueRequest协议消息
             * @param 时间戳（未使用）
             */
            void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("声明队列时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("声明队列时，没有找到信道！");
                    return;
                }
                return cp->declareQueue(message);
            }
            
            /**
             * @brief 处理删除队列请求
             * @param conn TCP连接对象
             * @param message deleteQueueRequest协议消息
             * @param 时间戳（未使用）
             */
            void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("删除队列时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("删除队列时，没有找到信道！");
                    return;
                }
                return cp->deleteQueue(message);
            }
            
            /**
             * @brief 处理队列绑定请求
             * @param conn TCP连接对象
             * @param message queueBindRequest协议消息
             * @param 时间戳（未使用）
             */
            void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("队列绑定时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("队列绑定时，没有找到信道！");
                    return;
                }
                return cp->queueBind(message);
            }
            
            /**
             * @brief 处理队列解绑请求
             * @param conn TCP连接对象
             * @param message queueUnBindRequest协议消息
             * @param 时间戳（未使用）
             */
            void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("队列解除绑定时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("队列解除绑定时，没有找到信道！");
                    return;
                }
                return cp->queueUnBind(message);
            }
            
            /**
             * @brief 处理消息发布请求
             * @param conn TCP连接对象
             * @param message basicPublishRequest协议消息
             * @param 时间戳（未使用）
             */
            void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("发布消息时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("发布消息时，没有找到信道！");
                    return;
                }
                return cp->basicPublish(message);
            }
            
            /**
             * @brief 处理消息确认请求
             * @param conn TCP连接对象
             * @param message basicAckRequest协议消息
             * @param 时间戳（未使用）
             */
            void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("确认消息时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("确认消息时，没有找到信道！");
                    return;
                }
                return cp->basicAck(message);
            }
            
            /**
             * @brief 处理消息消费订阅请求
             * @param conn TCP连接对象
             * @param message basicConsumeRequest协议消息
             * @param 时间戳（未使用）
             */
            void onBasicConsume(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("队列消息订阅时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("队列消息订阅时，没有找到信道！");
                    return;
                }
                return cp->basicConsume(message);
            }
            
            /**
             * @brief 处理取消消息订阅请求
             * @param conn TCP连接对象
             * @param message basicCancelRequest协议消息
             * @param 时间戳（未使用）
             */
            void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message, muduo::Timestamp) {
                Connection::ptr mconn = _connection_manager->getConnection(conn);
                if (mconn.get() == nullptr) 
                {
                    DLOG("队列消息取消订阅时，没有找到连接对应的Connection对象！");
                    conn->shutdown();
                    return;
                }
                Channel::ptr cp = mconn->getChannel(message->cid());
                if (cp.get() == nullptr) 
                {
                    DLOG("队列消息取消订阅时，没有找到信道！");
                    return;
                }
                return cp->basicCancel(message);
            }
            
            /**
             * @brief 处理未知协议消息
             * @param conn TCP连接对象
             * @param message 未知协议消息
             * @param 时间戳（未使用）
             * 
             * 记录未知消息类型并关闭连接
             */
            void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
                LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
                conn->shutdown();
            }
            
            /**
             * @brief TCP连接状态变化回调
             * @param conn TCP连接对象
             * 
             * 当连接建立时创建Connection对象，连接断开时销毁Connection对象
             */
            void onConnection(const muduo::net::TcpConnectionPtr &conn) 
            {
                if (conn->connected()) 
                {
                    _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);
                }
                else 
                {
                    _connection_manager->delConnection(conn);
                }
            }
            
        private:
            muduo::net::EventLoop _baseloop;          ///< 基础事件循环
            muduo::net::TcpServer _server;            ///< TCP服务器对象
            ProtobufDispatcher _dispatcher;           ///< Protobuf请求分发器
            ProtobufCodecPtr _codec;                  ///< Protobuf编解码器
            VirtualHost::ptr _virtual_host;           ///< 虚拟主机管理对象
            ConsumerManager::ptr _consumer_manager;   ///< 消费者管理对象
            ConnectionManager::ptr _connection_manager; ///< 连接管理对象
            threadpool::ptr _threadpool;              ///< 线程池对象
    };
}
#endif